package com.atguigu.realtime.sink;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.realtime.bean.TableProcess;
import com.atguigu.realtime.common.Constant;
import com.atguigu.realtime.util.RedisUtil;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import redis.clients.jedis.Jedis;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.SQLException;

/**
 * Created by RUI on 2021/6/27 14:05
 */
public class PhoenixSink extends RichSinkFunction<Tuple2<JSONObject, TableProcess>> {

    private Connection conn;
    private ValueState<Boolean> createTableState;

    @Override
    public void open(Configuration parameters) throws Exception {
        Class.forName(Constant.PHOENIX_DRIVER);
        conn = DriverManager.getConnection(Constant.PHOENIX_URL);

        createTableState = getRuntimeContext().getState(new ValueStateDescriptor<Boolean>("createTableState", Boolean.class));

    }

    @Override
    public void invoke(Tuple2<JSONObject, TableProcess> value, Context context) throws Exception {
        // 1.检测表是否存在, 如果不存在需要先建表
        checkTable(value);

        // 2.把这条数据写入到Phoenix中
        write2Hbase(value);

        // 3. dwm层需要的: 更新redis中的缓存
        // 如果检测到这次是update操作, 则需要去缓存中更新对应的维度数据或者删除对应的数据
        delCache(value);
    }

    private void delCache(Tuple2<JSONObject, TableProcess> value) {
        String key = value.f1.getSinkTable().toUpperCase() + ":" + value.f0.get("id");

        Jedis jedis = RedisUtil.getRedisClient();
        jedis.del(key);  // 如果key不存在, 会有啥影响?  没有影响. 所以不用提前判断key是否存在
        jedis.close();  // 归还给连接池
    }

    private void write2Hbase(Tuple2<JSONObject, TableProcess> value) throws SQLException {
        JSONObject data = value.f0;
        TableProcess tp = value.f1;

        StringBuilder sql = new StringBuilder();
        String[] columns = tp.getSinkColumns().split(",");

        sql
            .append("upsert into ")
            .append(tp.getSinkTable())
            .append("(")
            .append(tp.getSinkColumns())
            .append(")values(");

        // upsert into user(id, name, age) values(?,?,?)
        sql.append(tp.getSinkColumns().replaceAll("[^,]+", "?"));  // 正则替换

        sql.append(")");

        PreparedStatement ps = conn.prepareStatement(sql.toString());
        //给占位符赋值
        for (int i = 0; i < columns.length; i++) {

            String v = data.get(columns[i]) == null ? "" : data.get(columns[i]).toString();//避免空指针

            ps.setString(i+1,v);
        }

        ps.execute();
        conn.commit();
        ps.close();
    }

    private void checkTable(Tuple2<JSONObject, TableProcess> value) throws Exception {
        // 执行建表语句,实现在Phoenix中完成建表
        if (createTableState.value() == null) { // 可以避免每来一条数据都需要去执行一次sql
            TableProcess tp = value.f1;
            StringBuilder sql = new StringBuilder();
            sql
                .append("create table if not exists ")
                .append(tp.getSinkTable())
                .append("(");

            //example columns : id,coupon_id,range_type,range_id
            sql
                .append(tp.getSinkColumns().replaceAll(",", " varchar,"))
                .append(" varchar");

            //拼主键
            sql
                .append(", constraint pk primary key(")
                .append(tp.getSinkPk()==null?"id":tp.getSinkPk())
                .append("))");

            //加 盐表（hbase region分区个数）
            sql.append(tp.getSinkExtend()==null?"":tp.getSinkExtend());

            System.out.println("建表语句: " + sql.toString());

            PreparedStatement ps = conn.prepareStatement(sql.toString());
            ps.execute();
            conn.commit();
            ps.close();

            createTableState.update(true);
        }
    }

    @Override
    public void close() throws Exception {
        if (conn != null) {
            conn.close();
        }
    }
}
